package com.joizhang.seckill.service;

import com.joizhang.seckill.api.kafka.SecKillMessage;
import com.joizhang.seckill.api.redis.GoodsKey;
import com.joizhang.seckill.api.redis.SecKillKey;
import com.joizhang.seckill.api.service.SecKillService;
import com.joizhang.seckill.api.vo.*;
import com.joizhang.seckill.kafka.MQSender;
import com.joizhang.seckill.mapper.SecKillGoodsMapper;
import com.joizhang.seckill.model.SecKillGoods;
import com.joizhang.seckill.redis.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.util.Objects;

@DubboService
@Service
@Slf4j
public class SecKillServiceImpl implements SecKillService {

    public static final String HASH_KEY_STOCK_COUNT = "stockCount";
    /**
     * 乐观锁冲突最大重试次数
     */
    private static final int DEFAULT_MAX_RETRIES = 5;

    private final SecKillGoodsMapper secKillGoodsMapper;

    private final GoodsServiceImpl goodsService;

    private final OrderServiceImpl orderService;

    private final RedisService redisService;

    private final MQSender mqSender;

    private DefaultRedisScript<Long> script;

    @Autowired
    public SecKillServiceImpl(SecKillGoodsMapper secKillGoodsMapper, GoodsServiceImpl goodsService,
                              OrderServiceImpl orderService, RedisService redisService, MQSender mqSender) {
        this.secKillGoodsMapper = secKillGoodsMapper;
        this.goodsService = goodsService;
        this.orderService = orderService;
        this.redisService = redisService;
        this.mqSender = mqSender;
    }

    @PostConstruct
    public void init() {
        script = new DefaultRedisScript<>();
        script.setResultType(Long.class);
        script.setScriptSource(new ResourceScriptSource(new ClassPathResource("lua/decrementStock.lua")));
    }

    @Override
    public Boolean sendSecKillMessage(SecKillMessage message) {
        return mqSender.sendSecKillMessage(message);
    }

    @Override
    public void handleSecKillMessage(SecKillMessage secKillMessage) {
        UserVo user = secKillMessage.getUser();
        Long goodsId = secKillMessage.getGoodsId();
        // 判断秒杀结束
        SecKillGoodsVo secKillGoodsVo = goodsService.getSecKillGoodsVoByGoodsId(goodsId);
        if (secKillGoodsVo.getStockCount() <= 0) {
            return;
        }
        // 判断重复秒杀
        Integer orderCount = orderService.countOrderByUserIdGoodsId(user.getId(), goodsId);
        if (orderCount != 0) {
            return;
        }
        // 减库存 -> 下订单 -> 写入秒杀订单
        this.secKill(user, secKillGoodsVo);
    }

    @Override
    public Long decrement(GoodsKey getGoodsStock, String key) {
        return redisService.decrement(getGoodsStock, key);
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void secKill(UserVo user, SecKillGoodsVo secKillGoodsVo) {
        // 减秒杀商品库存
        boolean success = this.reduceStock(secKillGoodsVo);
        if (success) {
            // TODO 减总库存

            // 通过Lua脚本更新秒杀商品缓存
            redisService.execute(script, GoodsKey.GOODS_DETAIL,
                    secKillGoodsVo.getGoodsId().toString(), HASH_KEY_STOCK_COUNT, 1);

            // 写入秒杀订单
            GoodsVo goodsVo = goodsService.getGoodsVoByGoodsId(secKillGoodsVo.getGoodsId());
            orderService.createOrder(user, goodsVo, secKillGoodsVo);
        } else {
            setGoodsOver(secKillGoodsVo.getId());
        }
    }

    private boolean reduceStock(SecKillGoodsVo secKillGoodsVo) {
        int numAttempts = 0;
        int ret = 0;
        SecKillGoods secKillGoods = new SecKillGoods();
        BeanUtils.copyProperties(secKillGoodsVo, secKillGoods);
        do {
            numAttempts++;
            try {
                ret = secKillGoodsMapper.reduceStockByVersion(secKillGoods);
            } catch (Exception e) {
                log.error(e.getMessage());
            }
            if (ret != 0) {
                break;
            }
        } while (numAttempts < DEFAULT_MAX_RETRIES);

        return ret > 0;
    }

    @Override
    public Long getSecKillResult(UserVo userVo, Long goodsId) {
        OrderVo order = orderService.getOrderFromCache(userVo, goodsId);
        if (Objects.isNull(order)) {
            boolean isOver = getGoodsOver(goodsId);
            if (isOver) {
                return -1L;
            } else {
                return 0L;
            }
        } else {
            return order.getOrderInfoId();
        }
    }

    private void setGoodsOver(Long goodsId) {
        redisService.set(SecKillKey.isGoodsOver, goodsId.toString(), true);
    }

    private boolean getGoodsOver(Long goodsId) {
        return redisService.exists(SecKillKey.isGoodsOver, goodsId.toString());
    }

}
